/**
 * @file snapshot_synthesizer.cpp
 * @brief 快照合成器实现
 * @details 实现了快照合成器的功能，包括处理增量市场数据更新、维护限价订单簿快照和发布完整快照
 */

#include "snapshot_synthesizer.h"

namespace Exchange {
  /**
   * @namespace Exchange
   * @brief 交易所相关功能的命名空间
   */
  /**
   * @brief 快照合成器的构造函数
   * @param market_updates 市场数据发布器提供的增量市场数据更新队列
   * @param iface 网络接口名称
   * @param snapshot_ip 快照数据多播IP地址
   * @param snapshot_port 快照数据多播端口
   */
  SnapshotSynthesizer::SnapshotSynthesizer(MDPMarketUpdateLFQueue *market_updates, const std::string &iface,
                                           const std::string &snapshot_ip, int snapshot_port)
      : snapshot_md_updates_(market_updates), logger_("exchange_snapshot_synthesizer.log"), snapshot_socket_(logger_), order_pool_(ME_MAX_ORDER_IDS) {
    // 初始化快照数据多播套接字
    ASSERT(snapshot_socket_.init(snapshot_ip, iface, snapshot_port, /*is_listening*/ false) >= 0,
           "Unable to create snapshot mcast socket. error:" + std::string(std::strerror(errno)));
    // 初始化订单数组，将所有指针设置为nullptr
    for(auto& orders : ticker_orders_)
      orders.fill(nullptr);
  }

  /**
   * @brief 快照合成器的析构函数
   * @details 停止快照合成器线程
   */
  SnapshotSynthesizer::~SnapshotSynthesizer() {
    stop();
  }

  /**
   * @brief 启动快照合成器线程
   * @details 设置运行标志并创建新线程执行run方法
   */
  void SnapshotSynthesizer::start() {
    // 设置运行标志为true
    run_ = true;
    // 创建并启动快照合成器线程
    ASSERT(Common::createAndStartThread(-1, "Exchange/SnapshotSynthesizer", [this]() { run(); }) != nullptr,
           "Failed to start SnapshotSynthesizer thread.");
  }

  /**
   * @brief 停止快照合成器线程
   * @details 将运行标志设置为false，使线程退出循环
   */
  void SnapshotSynthesizer::stop() {
    run_ = false;
  }

  /**
   * @brief 处理增量市场数据更新并更新限价订单簿快照
   * @param market_update 要处理的市场数据更新
   * @details 根据更新类型（添加、修改、取消）更新内部限价订单簿快照
   */
  auto SnapshotSynthesizer::addToSnapshot(const MDPMarketUpdate *market_update) {
    // 获取市场数据更新的内容
    const auto &me_market_update = market_update->me_market_update_;
    // 获取对应证券ID的订单数组
    auto *orders = &ticker_orders_.at(me_market_update.ticker_id_);
    // 根据市场数据更新类型处理
    switch (me_market_update.type_) {
      // 处理添加订单的更新
      case MarketUpdateType::ADD: {
        // 检查该订单ID是否已存在
        auto order = orders->at(me_market_update.order_id_);
        // 断言订单不存在，否则报错
        ASSERT(order == nullptr, "Received:" + me_market_update.toString() + " but order already exists:" + (order ? order->toString() : ""));
        // 从内存池分配新订单并存储
        orders->at(me_market_update.order_id_) = order_pool_.allocate(me_market_update);
      }
        break;
      // 处理修改订单的更新
      case MarketUpdateType::MODIFY: {
        // 获取现有订单
        auto order = orders->at(me_market_update.order_id_);
        // 断言订单存在，否则报错
        ASSERT(order != nullptr, "Received:" + me_market_update.toString() + " but order does not exist.");
        // 断言订单ID匹配
        ASSERT(order->order_id_ == me_market_update.order_id_, "Expecting existing order to match new one.");
        // 断言交易方向匹配
        ASSERT(order->side_ == me_market_update.side_, "Expecting existing order to match new one.");

        // 更新订单的数量和价格
        order->qty_ = me_market_update.qty_;
        order->price_ = me_market_update.price_;
      }
        break;
      // 处理取消订单的更新
      case MarketUpdateType::CANCEL: {
        // 获取现有订单
        auto order = orders->at(me_market_update.order_id_);
        // 断言订单存在，否则报错
        ASSERT(order != nullptr, "Received:" + me_market_update.toString() + " but order does not exist.");
        // 断言订单ID匹配
        ASSERT(order->order_id_ == me_market_update.order_id_, "Expecting existing order to match new one.");
        // 断言交易方向匹配
        ASSERT(order->side_ == me_market_update.side_, "Expecting existing order to match new one.");

        // 释放订单内存并将指针设置为nullptr
        order_pool_.deallocate(order);
        orders->at(me_market_update.order_id_) = nullptr;
      }
        break;
      // 其他类型的市场数据更新不需要特殊处理
      case MarketUpdateType::SNAPSHOT_START:
      case MarketUpdateType::CLEAR:
      case MarketUpdateType::SNAPSHOT_END:
      case MarketUpdateType::TRADE:
      case MarketUpdateType::INVALID:
        break;
    }

    // 断言序列号连续增长，确保没有丢失消息
    ASSERT(market_update->seq_num_ == last_inc_seq_num_ + 1, "Expected incremental seq_nums to increase.");
    // 更新最后处理的序列号
    last_inc_seq_num_ = market_update->seq_num_;
  }

  /**
   * @brief 在快照多播流上发布完整的快照周期
   * @details 发布包含开始标记、清除指令、所有活跃订单和结束标记的完整快照
   */
  auto SnapshotSynthesizer::publishSnapshot() {
    // 初始化快照大小计数器
    size_t snapshot_size = 0;

    // 快照周期从一个SNAPSHOT_START消息开始，order_id_包含用于构建此快照的增量市场数据流的最后序列号
    // 创建快照开始消息
    const MDPMarketUpdate start_market_update{snapshot_size++, {MarketUpdateType::SNAPSHOT_START, last_inc_seq_num_}};
    // 记录日志
    logger_.log("%:% %() % %\n", __FILE__, __LINE__, __FUNCTION__, getCurrentTimeStr(&time_str_), start_market_update.toString());
    // 发送快照开始消息
    snapshot_socket_.send(&start_market_update, sizeof(MDPMarketUpdate));

    // 为每个证券的限价订单簿中的每个订单发布订单信息
    // 遍历所有证券
    for (size_t ticker_id = 0; ticker_id < ticker_orders_.size(); ++ticker_id) {
      // 获取当前证券的订单数组
      const auto &orders = ticker_orders_.at(ticker_id);

      // 创建清除消息
      MEMarketUpdate me_market_update;
      me_market_update.type_ = MarketUpdateType::CLEAR;
      me_market_update.ticker_id_ = ticker_id;

      // 我们首先发布一个CLEAR消息，以便下游消费者可以清除订单簿
      // 创建清除市场更新消息
      const MDPMarketUpdate clear_market_update{snapshot_size++, me_market_update};
      // 记录日志
      logger_.log("%:% %() % %\n", __FILE__, __LINE__, __FUNCTION__, getCurrentTimeStr(&time_str_), clear_market_update.toString());
      // 发送清除消息
      snapshot_socket_.send(&clear_market_update, sizeof(MDPMarketUpdate));

      // 发布每个订单
      // 遍历所有订单
      for (const auto order: orders) {
        // 只发布非空订单
        if (order) {
          // 创建市场更新消息
          const MDPMarketUpdate market_update{snapshot_size++, *order};
          // 记录日志
          logger_.log("%:% %() % %\n", __FILE__, __LINE__, __FUNCTION__, getCurrentTimeStr(&time_str_), market_update.toString());
          // 发送订单信息
          snapshot_socket_.send(&market_update, sizeof(MDPMarketUpdate));
          // 处理可能的接收数据
          snapshot_socket_.sendAndRecv();
        }
      }
    }

    // 快照周期以一个SNAPSHOT_END消息结束，order_id_包含用于构建此快照的增量市场数据流的最后序列号
    // 创建快照结束消息
    const MDPMarketUpdate end_market_update{snapshot_size++, {MarketUpdateType::SNAPSHOT_END, last_inc_seq_num_}};
    // 记录日志
    logger_.log("%:% %() % %\n", __FILE__, __LINE__, __FUNCTION__, getCurrentTimeStr(&time_str_), end_market_update.toString());
    // 发送快照结束消息
    snapshot_socket_.send(&end_market_update, sizeof(MDPMarketUpdate));
    // 处理可能的接收数据
    snapshot_socket_.sendAndRecv();

    // 记录发布快照的数量统计信息
    logger_.log("%:% %() % Published snapshot of % orders.\n", __FILE__, __LINE__, __FUNCTION__, getCurrentTimeStr(&time_str_), snapshot_size - 1);
  }

  /**
   * @brief 线程主方法
   * @details 处理来自市场数据发布器的增量更新，更新快照并定期发布快照
   */
  void SnapshotSynthesizer::run() {
    // 记录线程启动日志
    logger_.log("%:% %() %\n", __FILE__, __LINE__, __FUNCTION__, getCurrentTimeStr(&time_str_));
    // 主循环，当run_为true时持续运行
    while (run_) {
      // 遍历所有待处理的市场数据更新
      for (auto market_update = snapshot_md_updates_->getNextToRead(); snapshot_md_updates_->size() && market_update; market_update = snapshot_md_updates_->getNextToRead()) {
        // 记录正在处理的市场数据更新
        logger_.log("%:% %() % Processing %\n", __FILE__, __LINE__, __FUNCTION__, getCurrentTimeStr(&time_str_),
                    market_update->toString().c_str());

        // 将市场数据更新添加到快照中
        addToSnapshot(market_update);

        // 更新读取索引，表示已处理完毕当前市场数据更新
        snapshot_md_updates_->updateReadIndex();
      }

      // 检查是否需要发布快照（每60秒发布一次）
      if (getCurrentNanos() - last_snapshot_time_ > 60 * NANOS_TO_SECS) {
        // 更新最后发布快照的时间
        last_snapshot_time_ = getCurrentNanos();
        // 发布快照
        publishSnapshot();
      }
    }
  }
}
